package com.ly.kafka.controller;

import com.ly.kafka.empty.Order;
import com.ly.kafka.empty.OrderDTO;
import com.ly.kafka.service.KafkaService;
import com.ly.kafka.utils.GsonUtil;
import com.ly.kafka.utils.ProducerUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

@RestController
@RequestMapping("/kafka")
public class KafkaController {

    @Resource
    private KafkaTemplate kafkaTemplate;

    @Resource
    private ProducerUtils producerUtils;

    @Autowired
    private KafkaService kafkaService;


    /**
     * @describe：kafka声明事务
     * @author：李勇
     * @date：2023/4/11
     * @param
     * @return：void
     */
    @GetMapping("/transaction")
    public void sendMessage7(){
        // 声明事务：后面报错消息不会发出去
        kafkaTemplate.executeInTransaction(operations -> {
            operations.send("topic1","test executeInTransaction");
            throw new RuntimeException("fail");
        });

        // 不声明事务：后面报错但前面消息已经发送成功了
        kafkaTemplate.send("topic1","test executeInTransaction");
        throw new RuntimeException("fail");
    }


    @GetMapping(value = "/test1")
    public String test1(){
        try {
            String key = "hello-kraft1-1";
            Integer partition = 0;
            producerUtils.sendMessage("hello-kraft1", partition, key, new Order(1,23,12,24.0));
            return "发送成功";
        }catch (Exception e){
            e.printStackTrace();
            return "发送失败";
        }
    }

    @GetMapping(value = "/test2")
    public String test2(){
        try {
            String key = "order-topic-test-1";
            Integer partition = 0;
            producerUtils.sendMessage("order-topic-test", partition, key, new Order(1,23,12,24.0));
            return "发送成功";
        }catch (Exception e){
            e.printStackTrace();
            return "发送失败";
        }
    }

    @GetMapping(value = "/test3")
    public String test3(){
        try {
            String key = "order-topic-test-2";
            Integer partition = 0;
            producerUtils.sendMessage("order-topic-test-2", partition, key,3);
            return "发送成功";
        }catch (Exception e){
            e.printStackTrace();
            return "发送失败";
        }
    }

    @GetMapping(value = "/test4")
    public String test4(){
        try {
            String key = "hello-kraft3";
            Integer partition = 0;
            producerUtils.sendMessage("hello-kraft3", partition, key,3);
            return "发送成功";
        }catch (Exception e){
            e.printStackTrace();
            return "发送失败";
        }
    }


    @GetMapping(value = "/test5")
    public String test5(){
        try {
            String key = "hello-kraft";

                producerUtils.sendMessage(key, 0, key,"测试kafka广播");
            return "发送成功";
        }catch (Exception e){
            e.printStackTrace();
            return "发送失败";
        }
    }

    @GetMapping(value = "/test6")
    public String test6(){
        try {
            for (long i = 0; i < 10; i++) {
                OrderDTO order = new OrderDTO();
                order.setOrderName("订单"+i);
                order.setId(i);
                producerUtils.sendMessage("baiyan-test1", 0, "baiyan-test1",GsonUtil.gsonToString(order));
            }
            return "发送成功";
        }catch (Exception e){
            e.printStackTrace();
            return "发送失败";
        }
    }







    @GetMapping(value = "/editTopicPartitionNum")
    public void editTopicPartitionNum(){
        kafkaService.editTopicPartitionNum("baiyan-test2",2);
    }

}
